package hang.kafka.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @author 航
 * @date 2024/04/14 12:31
 **/
public class ProducerCallbackTest {
    public static void main(String[] args) {
        // 创建 配置对象
        Map<String, Object> config = new HashMap<String, Object>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 对生产者的数据 K，V 进行序列化操作
        // StringSerializer.class.getName() 表示获取这个类的名称
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());


        // 创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer(config);

//        // 创建数据
//        // 构建数据，需要传递三个参数：topicName，数据的key，数据的value
//        ProducerRecord<String, String> record = new ProducerRecord("test","key","value");
//
//        // 通过生产者对象发送数据
//        producer.send(record);


        /**
         * kafka应答级别有三个（ACKS）：-1(all),0,1
         * 默认是 -1  ：这个级别会等Broker中所有isSync的副本全部将数据保存到磁盘中去，才会响应
         * 0：这个级别会等Broker（Leader）将数据保存到磁盘中去，才会响应
         * 1：这个级别只要kafka的Producer将数据发送完给Broker（Leader）就会响应，发完就行，不用等那边给接受指令
         */
        for(int i = 0; i<10; i++){
            ProducerRecord<String, String> record = new ProducerRecord("test","key" + i,"value" + i);
            // 异步发送
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if(e == null){
                        System.out.println("数据发送成功：" + recordMetadata);
                    }
                }
            });
        }

        for(int i = 0; i<10; i++){
            ProducerRecord<String, String> record = new ProducerRecord("test","key" + i,"value" + i);
            // 同步发送
            Future<RecordMetadata> send = producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null) {
                        System.out.println("数据发送成功：" + recordMetadata);
                    }
                }
            });
            System.out.println("数据发送");
            // 同步发送
            try {
                send.get();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } catch (ExecutionException e) {
                throw new RuntimeException(e);
            }
        }

        // 关闭生产者对象
        producer.close();
    }
}
